package com.cwz.mq.ack.config;

import com.cwz.core.constant.MqConstants;
import com.cwz.mq.ack.constant.MsgLogConstants;
import com.cwz.mq.ack.service.MsgLogService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * @program: w-demo
 * @description: rabbitmq 配置类（配置 rabbitTemplate、初始化队列、交换机等）
 * @author: Wen
 **/
@Configuration
@Slf4j
public class RabbitConfig {
	@Autowired
	private CachingConnectionFactory cachingConnectionFactory;
	@Autowired
	private MsgLogService msgLogService;

	@Bean
	public RabbitTemplate rabbitTemplate() {
		// 发送消息确认：用来确认生产者 producer 将消息发送到 broker ，broker 上的交换机 exchange 再投递给队列 queue 的过程中，消息是否成功投递。
		RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);
		rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());

		// 消息从 producer 到 rabbitmq broker有一个 confirmCallback 确认模式。
		// correlationData：对象内部只有一个 id 属性，用来表示当前消息的唯一性。
		// ack：消息投递到 broker 的状态，true 表示成功。
		// cause：表示投递失败的原因。
		rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
			if (ack) {
				this.msgLogService.updateMsgStatus(correlationData.getId(), MsgLogConstants.DELIVER_SUCCESS);
			} else {
				log.error("消息发送到 Exchange 失败, {}, cause: {}", correlationData, cause);
				this.msgLogService.updateMsgStatus(correlationData.getId(), MsgLogConstants.DELIVER_FAIL);
			}
		});

		// 触发 setReturnCallback 回调必须设置 mandatory = true, 否则 Exchange 没有找到 Queue 就会丢弃掉消息, 而不会触发回调
		// 开启强制委托模式
		rabbitTemplate.setMandatory(true);
		// 消息从 exchange 到 queue 投递失败有一个 returnCallback 退回模式。
		rabbitTemplate.setReturnsCallback(rMsg -> {
			log.info("消息从 Exchange 到 Queue 失败: exchange: {}, route: {}, replyCode: {}, replyText: {}, message: {}",
				rMsg.getExchange(), rMsg.getRoutingKey(), rMsg.getReplyCode(), rMsg.getReplyText(), rMsg.getMessage());
			this.msgLogService.updateMsgStatus(RabbitConfig.getCorrelationId(rMsg.getMessage()), MsgLogConstants.DELIVER_FAIL);
		});
		return rabbitTemplate;
	}

	/**
	 * @Description: 获取 correlationId
	 * @Author: Wen
	 * @param message:
	 * @return: java.lang.String
	 **/
	public static String getCorrelationId(Message message) {
		String correlationId = null;
		for (Map.Entry entry : message.getMessageProperties().getHeaders().entrySet()) {
			String key = (String) entry.getKey();
			String value = (String) entry.getValue();
			if (key.equals("spring_returned_message_correlation")) {
				correlationId = value;
				break;
			}
		}
		return correlationId;
	}

	// 1. FanoutExchange (扇型交换机)-  全部路由
	// 广播模式或者订阅模式，可以同时绑定多个 queue, 发送消息时，无需指定 Routing Key
	// 适用场景：广播，群聊，新闻推送
	// 2. DirectExchange（直连交换机）- 根据 Routing Key 路由
	// 直连，通过 Routing Key 绑定 queue，当发送消息到交换机时，会根据配置的 Routing Key 路由到不同的 queue 中
	// 当 N 个 queue 的 Routing Key 相同时，消息会被同时路由到这 N 个 queue 中去
	// 3. HeadersExchange（头交换机）- 根据 header 的匹配规则路由
	// 设置交换机的匹配 header 的规则，支持单个精确匹配 where，部分匹配 whereAny，全部匹配 whereAll，根据匹配的结果路由到相应的 queue 中
	// 4. TopicExchange（主题交换机）
	// 主题交换机通过有匹配规则的路由键和队列绑定，*.a.*，#.b.#，* 代表匹配任意一个单词，# 代表匹配任意一个或多个单词
	// 发送消息时，设置路由键，如设置为 l.a.m，则会被路由到绑定 *.a.* 的队列去，设置为 l.b 则会被路由到绑定 #.b.# 的队列去，
	// 设置为 l.a.b.c 会同时到两个队列中，设置 l.m 则会被丢弃
	// 5. CustomExchange 不是一种固定的类型，是用来配合插件一起使用的，比如延迟队列

	/**
	 * 测试 mq
	 */
	@Bean
	public Queue sendTestQueue() {
		return new Queue(MqConstants.SEND_TEST_QUEUE, true);
	}

	@Bean
	public DirectExchange sendTestExchange() {
		return new DirectExchange(MqConstants.SEND_TEST_EXCHANGE, true, false);
	}

	@Bean
	public Binding sendTestBinding(@Qualifier("sendTestQueue") Queue queue,
								   @Qualifier("sendTestExchange") DirectExchange exchange) {
		return BindingBuilder.bind(queue).to(exchange).with(MqConstants.SEND_TEST_ROUTING);
	}

	/**
	 * 延迟 mq
	 */
	@Bean
	public Queue sendDelayedQueue() {
		return new Queue(MqConstants.SEND_DELAYED_QUEUE, true);
	}

	@Bean
	public CustomExchange sendDelayedExchange() {
		Map<String, Object> args = new HashMap<>();
		args.put("x-delayed-type", "direct");
		return new CustomExchange(MqConstants.SEND_DELAYED_EXCHANGE, "x-delayed-message", true, false, args);
	}

	@Bean
	public Binding sendDelayedBinding(@Qualifier("sendDelayedQueue") Queue queue,
								      @Qualifier("sendDelayedExchange") CustomExchange exchange) {
		return BindingBuilder.bind(queue).to(exchange).with(MqConstants.SEND_DELAYED_ROUTING).noargs();
	}

}
